#
# This file is a part of the open-eBackup project.
# This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0.
# If a copy of the MPL was not distributed with this file, You can obtain one at
# http://mozilla.org/MPL/2.0/.
#
# Copyright (c) [2024] Huawei Technologies Co.,Ltd.
#
# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
#

import datetime
import os
import re
import time

import pexpect
import psutil

from common.common import execute_cmd, check_command_injection_exclude_quote
from common.common_models import ActionResult, LogDetail, SubJobModel
from common.const import ParamConstant, CMDResult, DBLogLevel, SubJobTypeEnum, SubJobPolicyEnum, ExecuteResultEnum, \
    AuthType
from common.exception.common_exception import ErrCodeException
from common.number_const import NumberConst
from common.util.check_user_utils import check_os_user, check_path_owner
from common.util.check_utils import is_port
from common.util.cmd_utils import cmd_format
from common.util.exec_utils import exec_overwrite_file
from gaussdbt.single import LOGGER
from gaussdbt.single.comm.gst_single_const import GstSingleConst
from gaussdbt.single.comm.gst_single_error_code import ErrorCode


class GstSingleCommonUtil:
    @staticmethod
    def check_params(check_users: list = None, check_paths: list = None, check_path_owners: list = None):
        check_users = check_users or list()
        check_paths = check_paths or list()
        check_path_owners = check_path_owners or list()
        for user in check_users:
            if not check_os_user(user):
                LOGGER.error("The os user: %s is invalid.", user)
                raise ErrCodeException(ErrorCode.PARAMS_IS_INVALID, message="The os user is invalid.")
        for path in check_paths:
            if check_command_injection_exclude_quote(path):
                LOGGER.error("The path: %s is invalid.", path)
                raise ErrCodeException(ErrorCode.PARAMS_IS_INVALID, message="The path is invalid.")
        for path, owners in check_path_owners:
            if not check_path_owner(path, owners):
                LOGGER.error("The owner of path: %s is not %s.", path, owners)
                raise ErrCodeException(ErrorCode.PARAMS_IS_INVALID, message="The owner of path not match.")

    @staticmethod
    def get_gsdb_port(os_user) -> str:
        gsdb_data = GstSingleCommonUtil.get_val_of_gsdb_env_var(os_user, "GSDB_DATA")
        zengine_ini = os.path.realpath(os.path.join(gsdb_data, "cfg", "zengine.ini"))
        GstSingleCommonUtil.check_params(check_paths=[zengine_ini], check_path_owners=[(zengine_ini, [os_user])])
        with open(zengine_ini, mode='r') as z_ini:
            all_lines = z_ini.readlines()
        for i in all_lines:
            match_ret = re.match(r'LSNR_PORT\s*=', i, flags=re.IGNORECASE)
            if not match_ret:
                continue
            db_port = i.split("=")[1].strip()
            if not is_port(db_port):
                LOGGER.error("The LSNR_PORT value(%s) is not port.", db_port)
                raise ErrCodeException(ErrorCode.FAILED_EXECUTE_COMMAND,
                                       *[f"cat {zengine_ini}", "The LSNR_PORT value is not port."])
            return db_port
        raise ErrCodeException(ErrorCode.FAILED_EXECUTE_COMMAND, *[f"cat {zengine_ini}", "Get LSNR_PORT param failed."])

    @staticmethod
    def get_val_of_gsdb_env_var(os_user, env_var_name) -> str:
        if not check_os_user(os_user):
            LOGGER.error("The os user: %s is invalid.", os_user)
            raise ErrCodeException(ErrorCode.PARAMS_IS_INVALID, message="The os user is invalid.")
        if env_var_name not in GstSingleConst.GST_ENV_VAR_WHITELIST:
            LOGGER.error("The gaussdbt environment variable %s is not supported.", env_var_name)
            raise ErrCodeException(ErrorCode.PARAMS_IS_INVALID, message="The environment variable is not supported.")
        echo_cmd = cmd_format("echo ${}", env_var_name)
        cmd = f"su - {os_user} -c '{echo_cmd}'"
        ret_code, out, err = execute_cmd(cmd)
        if ret_code != CMDResult.SUCCESS.value:
            LOGGER.error("Get gaussdbt environment variable(%s) fail, code: %s, out: %s, err: %s.",
                         env_var_name, ret_code, out, err)
            raise ErrCodeException(ErrorCode.FAILED_EXECUTE_COMMAND, *[cmd, out])
        LOGGER.info("Get gaussdbt environment variable(%s) success, value: %s.", env_var_name, out)
        return out.strip()

    @staticmethod
    def check_gsdb_status(
            os_user, gsdb_home, gsdb_data, enable_sysdba_login=True, sys_pwd=None):
        LOGGER.info("Start checking gaussdbt status, os user: %s, db home: %s, db data: %s, enable login: %s.",
                    os_user, gsdb_home, gsdb_data, enable_sysdba_login)
        zctl_path = os.path.realpath(os.path.join(gsdb_home, "bin", "zctl.py"))
        GstSingleCommonUtil.check_params(
            check_users=[os_user], check_paths=[zctl_path, gsdb_data], check_path_owners=[(zctl_path, [os_user])])
        if enable_sysdba_login:
            zctl_path_cmd = cmd_format("python3 {} -t status -D {}", zctl_path, gsdb_data)
            cmd = f"su - {os_user} -c '{zctl_path_cmd}'"
            ret_code, out, err = execute_cmd(cmd)
            if ret_code != CMDResult.SUCCESS.value:
                LOGGER.error("Check gaussdbt status fail, code: %s, out: %s, err: %s.", ret_code, out, err)
                raise ErrCodeException(ErrorCode.STATUS_ERROR, message="Execute check database status command failed.")
            if "Current instance STATUS OPEN" not in out:
                LOGGER.error("Check gaussdbt status is not open, result: %s.", out)
                raise ErrCodeException(ErrorCode.STATUS_ERROR, message="Current instance status not open.")
            LOGGER.info("Check gaussdbt status is open, result: %s.", out)
            return
        su_cmd = f"su - {os_user}"
        ssh = pexpect.spawn(su_cmd, encoding='utf-8')
        try:
            index = ssh.expect(['[#\$>]', pexpect.EOF, pexpect.TIMEOUT], timeout=NumberConst.TEN)
            if index != 0:
                LOGGER.error("Query gaussdbt status change user failed, index: %s.", index)
                raise ErrCodeException(ErrorCode.FAILED_EXECUTE_COMMAND, *[su_cmd, ssh.before])
            query_status_cmd = cmd_format("python3 {} -t status -D {} -P", zctl_path, gsdb_data)
            ssh.sendline(query_status_cmd)
            index = ssh.expect(['Username:', pexpect.EOF, pexpect.TIMEOUT], timeout=NumberConst.TEN)
            if index != 0:
                out = ssh.before
                LOGGER.error("Query gaussdbt status expect username failed, index: %s, out: %s.", index, out)
                raise ErrCodeException(ErrorCode.FAILED_EXECUTE_COMMAND, *[query_status_cmd, out])
            ssh.sendline(GstSingleConst.SYS_DBA)
            index = ssh.expect(['Password:', pexpect.EOF, pexpect.TIMEOUT], timeout=NumberConst.TEN)
            if index != 0:
                LOGGER.error("Query gaussdbt status expect password failed, index: %s.", index)
                raise ErrCodeException(ErrorCode.FAILED_EXECUTE_COMMAND, *[query_status_cmd, ssh.before])
            ssh.sendline(sys_pwd)
            index = ssh.expect(['[#\$>]', pexpect.EOF, pexpect.TIMEOUT], timeout=NumberConst.THIRTY)
            if index != 0:
                LOGGER.error("Query gaussdbt status execute command failed, index: %s.", index)
                raise ErrCodeException(ErrorCode.FAILED_EXECUTE_COMMAND, *[query_status_cmd, ssh.before])
            chk_ret = ssh.before
            if "Current instance STATUS OPEN" not in chk_ret:
                LOGGER.error("Check gaussdbt status is not open, result: %s.", chk_ret)
                raise ErrCodeException(ErrorCode.STATUS_ERROR, message="Current gaussdbt instance not open.")
            LOGGER.info("Check gaussdbt status is open, result: %s.", chk_ret)
        finally:
            if ssh:
                ssh.close()

    @staticmethod
    def check_gsdb_auth(
            os_user, gsdb_home, gsdb_data, sys_auth_type, sys_pwd=None):
        LOGGER.info("Start checking gaussdbt auth, os user: %s, db home: %s, db data: %s, enable login: %s.",
                    os_user, gsdb_home, gsdb_data, sys_auth_type)
        if sys_auth_type not in [str(AuthType.NO_AUTO.value), str(AuthType.APP_PASSWORD.value)]:
            LOGGER.error("Check gaussdbt sys auth type error, type: %s.", sys_auth_type)
            return False, GstSingleCommonUtil.build_action_result(code=ExecuteResultEnum.INTERNAL_ERROR.value,
                                                                  body_err=ErrorCode.CHECK_CONNECTIVITY_FAILED)
        GstSingleCommonUtil.check_params(check_users=[os_user], check_paths=[gsdb_home, gsdb_data])
        enable_sysdba_login = (str(sys_auth_type) != str(AuthType.APP_PASSWORD.value))
        if enable_sysdba_login:
            no_auth_cmd = f'su - {os_user} -c "zsql / as sysdba -q"'
            ret_code, out, err = execute_cmd(no_auth_cmd)
            if "Failed to establish uds connection to" in out:
                LOGGER.error(f"Execute login connection failed, database service abnormal.")
                return False, GstSingleCommonUtil.build_action_result(code=ExecuteResultEnum.INTERNAL_ERROR.value,
                                                                      body_err=ErrorCode.ERR_DATABASE_STATUS)
            if ret_code != CMDResult.SUCCESS.value:
                LOGGER.error("Check gaussdbt status fail, code: %s, out: %s, err: %s.", ret_code, out, err)
                return False, GstSingleCommonUtil.build_action_result(code=ExecuteResultEnum.INTERNAL_ERROR.value,
                                                                      body_err=ErrorCode.CHECK_CONNECTIVITY_FAILED)
            return True, GstSingleCommonUtil.build_action_result(code=ExecuteResultEnum.SUCCESS.value)
        db_port = GstSingleCommonUtil.get_gsdb_port(os_user)
        log_cmd = cmd_format("zsql {}@127.0.0.1:{} -q", GstSingleConst.SYS_DBA, db_port)
        auth_cmd = f'su - {os_user} -c "{log_cmd}"'
        ssh = pexpect.spawn(auth_cmd, encoding='utf-8')
        try:
            index = ssh.expect(['password:', pexpect.EOF, pexpect.TIMEOUT], timeout=NumberConst.TEN)
            if index != 0:
                LOGGER.error("Execute zsql login command failed, index: %s.", index)
                return False, GstSingleCommonUtil.build_action_result(code=ExecuteResultEnum.INTERNAL_ERROR.value,
                                                                      body_err=ErrorCode.CHECK_CONNECTIVITY_FAILED)
            ssh.sendline(sys_pwd)
            index = ssh.expect(['connected.', 'Failed to establish tcp connection to', pexpect.EOF, pexpect.TIMEOUT],
                               timeout=NumberConst.THIRTY)
            if index == 1:
                LOGGER.error(f"Execute login connection failed, database service abnormal.")
                return False, GstSingleCommonUtil.build_action_result(code=ExecuteResultEnum.INTERNAL_ERROR.value,
                                                                      body_err=ErrorCode.ERR_DATABASE_STATUS)
            if index != 0:
                LOGGER.error("Execute login connection failed, index: %s.", index)
                return False, GstSingleCommonUtil.build_action_result(code=ExecuteResultEnum.INTERNAL_ERROR.value,
                                                                      body_err=ErrorCode.LOGIN_FAILED)
            LOGGER.info("Execute sql command success.")
            return True, GstSingleCommonUtil.build_action_result(code=ExecuteResultEnum.SUCCESS.value)
        finally:
            if ssh:
                ssh.close()

    @staticmethod
    def get_gsdb_version(os_user):
        version = ''
        cmd = f'su - {os_user} -c "zctl.py -v"'
        ret_code, out, err = execute_cmd(cmd)
        if ret_code != CMDResult.SUCCESS.value:
            LOGGER.error("Get gaussdbt version failed.")
            return version
        version = str(out).strip()
        return version

    @staticmethod
    def check_gaussdbt_single_service(os_user):
        cmd = f'su - {os_user} -c "gs_om -t status"'
        ret_code, out, err = execute_cmd(cmd)
        return ret_code != CMDResult.SUCCESS.value

    @staticmethod
    def exec_gsdb_sql(os_user, sql, enable_sysdba_login=True, sys_pwd=None) -> str:
        if enable_sysdba_login:
            cmd = f"su - {os_user} -c " + '''"zsql / as sysdba -q -c \\"''' + sql + '''\\""'''
            LOGGER.info("Start executing command: %s.", cmd)
            ret_code, out, err = execute_cmd(cmd)
            if ret_code != CMDResult.SUCCESS.value:
                LOGGER.error("Execute sql command failed, code: %s, out: %s, err: %s.", ret_code, out, err)
                raise ErrCodeException(ErrorCode.FAILED_EXECUTE_COMMAND, *[cmd, out])
            LOGGER.info("Execute sql command success.")
            return out
        db_port = GstSingleCommonUtil.get_gsdb_port(os_user)
        zsql_cmd = f'zsql {GstSingleConst.SYS_DBA}@127.0.0.1:{db_port} -q -c \"{sql}\"'
        LOGGER.info("Start executing command: %s.", zsql_cmd)
        su_cmd = f"su - {os_user}"
        ssh = pexpect.spawn(su_cmd, encoding='utf-8')
        try:
            index = ssh.expect(['[#\$>]', pexpect.EOF, pexpect.TIMEOUT], timeout=NumberConst.TEN)
            if index != 0:
                LOGGER.error("Execute zsql change user failed, index: %s.", index)
                raise ErrCodeException(ErrorCode.FAILED_EXECUTE_COMMAND, *[su_cmd, ssh.before])
            ssh.sendline(zsql_cmd)
            index = ssh.expect(['password:', pexpect.EOF, pexpect.TIMEOUT], timeout=NumberConst.TEN)
            if index != 0:
                cmd_ret = ssh.before
                LOGGER.error("Execute zsql command failed, index: %s.", index)
                raise ErrCodeException(ErrorCode.FAILED_EXECUTE_COMMAND, *[zsql_cmd, cmd_ret])
            ssh.sendline(sys_pwd)
            index = ssh.expect(['rows fetched.', pexpect.EOF, pexpect.TIMEOUT], timeout=NumberConst.THIRTY)
            if index != 0:
                cmd_ret = ssh.before
                LOGGER.error("Execute sql statement failed, index: %s.", index)
                raise ErrCodeException(ErrorCode.FAILED_EXECUTE_COMMAND, *[zsql_cmd, cmd_ret])
            LOGGER.info("Execute sql command success.")
            return ssh.before
        finally:
            if ssh:
                ssh.close()

    @staticmethod
    def kill_agent_job_process(job_pid, job_id):
        """杀掉agent任务进程"""
        pid_list = psutil.pids()
        for pid in pid_list:
            process = psutil.Process(pid)
            try:
                p_user = process.username()
            except Exception:
                LOGGER.debug("This is a temporary process, get username failed, process id: %s.", pid)
                continue
            if p_user != "rdadmin":
                continue
            try:
                cmd = process.cmdline()
            except Exception:
                LOGGER.debug("This is a temporary process, get cmdline failed, process id: %s.", pid)
                continue
            if "python3" in cmd and job_id in cmd and job_pid not in cmd:
                try:
                    process.kill()
                except Exception:
                    LOGGER.exception("Kill agent job process failed, process id: %s.", pid)
                    break
                LOGGER.info("The agent job has been killed, process id: %s, cmd: %s, job pid: %s, job id: %s.",
                            pid, cmd, job_pid, job_id)
                break

    @staticmethod
    def convert_backup_time_to_timestamp(time_str):
        """
        将备份时间转为时间戳
        :param time_str: 备份时间
        :return: 时间戳
        """
        struct_time = time.strptime(time_str, GstSingleConst.GSDB_TIME_FORMAT)
        return int(time.mktime(struct_time))

    @staticmethod
    def convert_time_to_utc_time(local_time_str):
        """
        将当前时间转为UTC时间
        :param local_time_str: 本地时间
        :return: UTC时间
        """
        struct_time = time.strptime(local_time_str, GstSingleConst.GSDB_TIME_FORMAT)
        utc_date = datetime.datetime.utcfromtimestamp(int(time.mktime(struct_time)))
        return utc_date.strftime(GstSingleConst.GSDB_TIME_FORMAT)

    @staticmethod
    def build_sub_job(job_id, job_name, priority, node_id):
        return SubJobModel(jobId=job_id, subJobId="", jobType=SubJobTypeEnum.BUSINESS_SUB_JOB.value, jobName=job_name,
                           jobPriority=priority, policy=SubJobPolicyEnum.FIXED_NODE.value, ignoreFailed=False,
                           execNodeId=node_id, jobInfo="").dict(by_alias=True)

    @staticmethod
    def build_action_result(code, body_err=None, err_params: list = None, message=None):
        return ActionResult(code=code, bodyErr=body_err, message=message, bodyErrParams=err_params)

    @staticmethod
    def build_log_detail(log_label, log_info_params, err_code=None,
                         log_detail_param=None, log_level=DBLogLevel.INFO.value):
        """
        构建事件详情
        :param log_label: 事件标签
        :param log_info_params: 事件参数
        :param err_code: 错误码
        :param log_detail_param: 错误码参数
        :param log_level: 事件级别
        :return: 事件详情
        """
        log_info_params = log_info_params if log_info_params else list()
        return LogDetail(logInfo=log_label,
                         logInfoParam=log_info_params,
                         logTimestamp=int(time.time()),
                         logDetail=err_code,
                         logDetailParam=log_detail_param,
                         logLevel=log_level)

    @staticmethod
    def output_action_result(pid, code, body_err=None, body_err_params: list = None, message=None):
        """
        将ActionResult写入到结果文件
        :param pid: 任务PID
        :param code: 执行结果
        :param body_err: 错误码
        :param body_err_params: 错误码参数
        :param message: 错误信息
        :return:
        """
        action_result = ActionResult(code=code, bodyErr=body_err, bodyErrParams=body_err_params, message=message)
        LOGGER.info("Output action result: %s.", action_result.dict(by_alias=True))
        result_file_path = os.path.realpath(os.path.join(ParamConstant.RESULT_PATH, f"result{pid}"))
        exec_overwrite_file(result_file_path, action_result.dict(by_alias=True))
